package com.atguigu.userprofile.ml.pipeline

import org.apache.spark.ml.classification.{DecisionTreeClassificationModel, DecisionTreeClassifier}
import org.apache.spark.ml.feature.{IndexToString, StringIndexer, StringIndexerModel, VectorAssembler, VectorIndexer}
import org.apache.spark.ml.{Pipeline, PipelineModel, Transformer}
import org.apache.spark.mllib.evaluation.MulticlassMetrics
import org.apache.spark.sql.DataFrame

class MyPipeline {


  var pipeline:Pipeline=null;

  def init():MyPipeline={
    pipeline=new Pipeline().setStages(Array(
      createLabelIndexer(),
      createFeatureAssemble(),
      createFeatrueIndexer(),
      createClassifier()
    ))
    this
  }



  var pipelineModel:PipelineModel=null;

   var labelColName:String=null;

   def setLabelColName(labelColName:String): MyPipeline ={
     this.labelColName=labelColName
     this
   }
   var featureColNames:Array[String]=null;

  def setFeatureColNames(featureColNames:Array[String]): MyPipeline ={
    this.featureColNames=featureColNames
    this
  }

  var maxCategories:Int =10;
  def setMaxCategories(maxCategories:Int): MyPipeline ={
    this.maxCategories=maxCategories
    this
  }


  //// 以下为参数 ////////////////////

  // 最大分支数
  private var maxBins=24
  // 最大树深度
  private var maxDepth=5
  //最小分支包含数据条数
  private var minInstancesPerNode=1
  //最小分支信息增益
  private var minInfoGain=0.0


  def setMaxBins(maxBins:Int): MyPipeline ={
    this.maxBins=maxBins
    this
  }
  def setMaxDepth(maxDepth:Int): MyPipeline ={
    this.maxDepth=maxDepth
    this
  }

  def setMinInstancesPerNode(minInstancesPerNode:Int): MyPipeline ={
    this.minInstancesPerNode=minInstancesPerNode
    this
  }

  def setMinInfoGain(minInfoGain:Double): MyPipeline ={
    this.minInfoGain=minInfoGain
    this
  }



  // 1) 标签索引 ：
  //  专门整理标签数据（参考答案）
  //  1 提取标签值
   //  2 把标签值按照出现的比例，变为矢量值 （0,1,2,3....） 出现比例越高 ，矢量值越小
  //  参数  inputcol  : 原始标签列名称
    //        outputcol ： 新产生矢量列的名称
  def createLabelIndexer(): StringIndexer ={
        val indexer = new  StringIndexer()
        indexer.setInputCol(labelColName)
        indexer.setOutputCol("label_index")
        indexer
  }

   //2) 特征聚合
  //  1  提取特征
  //  2 把多列特征组合成为一个特征列
  //  参数  inputcols  : 原始特例列名称（多个）
  //        outputcol ： 聚合特征列
  def  createFeatureAssemble(): VectorAssembler ={
        val vectorAssembler = new VectorAssembler()
        vectorAssembler.setInputCols(featureColNames)
        vectorAssembler.setOutputCol("feature_assemble")

        vectorAssembler
  }

  //3)  特征索引
  //  把特征 向量化处理
  //   把原始值 转为向量值(0,1,2,3....) 出现比例越高 ，矢量值越小
  //   连续值特征不会别向量化处理 维持原值
  //  参数  inputcol   :  上一步的输出列
  //        outputcol ：  新建的特征向量列
  def createFeatrueIndexer(): VectorIndexer ={
      val vectorIndexer = new VectorIndexer()
    vectorIndexer.setInputCol("feature_assemble")
    vectorIndexer.setOutputCol("feature_index")
    vectorIndexer.setMaxCategories(maxCategories)  //区分是否为 连续值特征
    vectorIndexer.setHandleInvalid("skip") // error   //keep
    vectorIndexer
  }


  //4)  分类器
  //   决策树算法
  //
  def  createClassifier(): DecisionTreeClassifier ={
        val classifier = new DecisionTreeClassifier()
    classifier.setLabelCol("label_index")     // 标签向量列
    classifier.setFeaturesCol("feature_index")     //特征向量列
    classifier.setPredictionCol("prediction") //预测结果列
    classifier.setImpurity("gini")

    classifier.setMaxDepth(maxDepth)   //树深度
    classifier.setMinInstancesPerNode(minInstancesPerNode) //每个判断节点的最小样本数
    classifier.setMinInfoGain(minInfoGain)  //每个判断节点的最小信息收益 0 -0.5
    classifier.setMaxBins(maxBins)   //连续值 切分数


     // classifier 参数 可以优化
    classifier
  }


  //训练
  // 传入数据（含标签） + 流水线 ==> 流水线模型
  def train(dataFrame :DataFrame): Unit ={
    pipelineModel = pipeline.fit(dataFrame)

  }
  //预测
  def predict(dataFrame: DataFrame): DataFrame ={
    val predictedDataFrame: DataFrame = pipelineModel.transform(dataFrame)
    predictedDataFrame
  }

  // 打印决策树
  def  printTree(): Unit ={
    val transformer: Transformer = pipelineModel.stages(3)
    val classificationModel: DecisionTreeClassificationModel = transformer.asInstanceOf[DecisionTreeClassificationModel]
    println(classificationModel.toDebugString)


  }

  //为了更好的筛选特征  ，打印特征的权重
  def printFeatureWeight(): Unit ={
    val transformer: Transformer = pipelineModel.stages(3)
    val classificationModel: DecisionTreeClassificationModel = transformer.asInstanceOf[DecisionTreeClassificationModel]
    println(classificationModel.featureImportances)
  }

  def printEvaluateReport(predictedDF:DataFrame): Unit ={

    val predictionAndLabelRdd=predictedDF.rdd.map{row=>
      val prediction: Double = row.getAs[Double]("prediction")
      val label: Double = row.getAs[Double]("label_index")
      (prediction,label)
    }
    val metrics = new MulticlassMetrics(predictionAndLabelRdd)
    // 准确率（全体 ）   各个指标（男，女） 精确率 和召回率
    println(  s"总准确率：${metrics.accuracy}")
    val labels: Array[Double] = metrics.labels
    for (label <- labels ) {
      println(s" 矢量值为$label 的召回率：${metrics.recall(label)}")
      println(s" 矢量值为$label 的精确率：${metrics.precision(label)}")
    }


  }

   //把预测结果转为原值
  def convertOrigin(predictedDF:DataFrame): DataFrame ={
    //召唤大师兄
    val transformer: Transformer = pipelineModel.stages(0)
    val stringIndexerModel: StringIndexerModel = transformer.asInstanceOf[StringIndexerModel]
    val indexToString = new IndexToString()
    indexToString.setInputCol("prediction")
    indexToString.setOutputCol("prediction_origin")
    indexToString.setLabels(stringIndexerModel.labels) //放原值与矢量的对应关系    矢量=数组下标

    val convertedDF: DataFrame = indexToString.transform(predictedDF)
    convertedDF
  }

 //保存模型
  def saveModel(path:String): Unit ={
    pipelineModel.write.overwrite().save(path)
  }

  //加载模型
  def  loadModel(path:String): MyPipeline ={
    pipelineModel = PipelineModel.load(path)
    this
  }


}
